package com.kxn.multi.module.api.component.rocketmq.transaction;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.kxn.multi.module.service.dto.api.mq.TransactionMessageParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.nio.charset.StandardCharsets;

/**
 * rocketmq事务消息
 *
 *
 * 参考[https://github.com/apache/rocketmq-spring/wiki/%E4%BA%8B%E5%8A%A1%E6%B6%88%E6%81%AF]
 * 原理：https://rocketmq.apache.org/zh/docs/featureBehavior/04transactionmessage
 */
@Slf4j
@RocketMQTransactionListener
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    /**
     * 发送完办消息以后，执行本地事务,然后返回给broker消息状态
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {

        // ... local transaction process, return bollback, commit or unknown
        log.info("获取到的事务消息：{}", msg);

        byte[] body = (byte[]) msg.getPayload();
        // 转换为字符串
        String bodyString = new String(body, StandardCharsets.UTF_8);
        System.out.println("Received message body: " + bodyString);

        // 假设消息体是 JSON，转换为对象
        ObjectMapper objectMapper = new ObjectMapper();
        TransactionMessageParam param = null;
        try {
            param = objectMapper.readValue(bodyString, TransactionMessageParam.class);
        } catch (Exception e) {
            log.error("解析失败:{}", e.getMessage(), e);
        }
        log.info("获取到的事务消息内容-executeLocalTransaction：{}", JSONObject.toJSON(param));


        RocketMQLocalTransactionState rocketMQLocalTransactionState;
        switch (param.getStatus()) {
            case 0:
                rocketMQLocalTransactionState = RocketMQLocalTransactionState.UNKNOWN;
                break;
            case 1:
                rocketMQLocalTransactionState = RocketMQLocalTransactionState.COMMIT;
                break;
            case 2:
                rocketMQLocalTransactionState = RocketMQLocalTransactionState.ROLLBACK;
                break;
            default:
                rocketMQLocalTransactionState = RocketMQLocalTransactionState.UNKNOWN;
        }
        return rocketMQLocalTransactionState;
    }

    /**
     * 如果本地事务还没有执行完发送状态，在等待一段时间后，broker端回查事务状态；
     * 回查规则：
     * 本地事务执行完成后，若Broker端收到的本地事务返回状态为UNKNOW，或生产者应用退出导致本地事务未提交任何状态。
     * 则Broker端会向消息生产者发起事务回查，第一次回查后仍未获取到事务状态，则之后每隔一段时间会再次回查。
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // ... check transaction status and return bollback, commit or unknown
        log.info("获取到的事务消息-checkLocalTransaction：{}", msg);

        byte[] body = (byte[]) msg.getPayload();
        // 转换为字符串
        String bodyString = new String(body, StandardCharsets.UTF_8);
        System.out.println("Received message body: " + bodyString);

        // 假设消息体是 JSON，转换为对象
        ObjectMapper objectMapper = new ObjectMapper();
        TransactionMessageParam param = null;
        try {
            param = objectMapper.readValue(bodyString, TransactionMessageParam.class);
        } catch (Exception e) {
            log.error("解析失败:{}", e.getMessage(), e);
        }
        log.info("获取到的事务消息内容-checkLocalTransaction：{}", JSONObject.toJSON(param));

        RocketMQLocalTransactionState rocketMQLocalTransactionState;
        switch (param.getStatus()) {
            case 0:
                rocketMQLocalTransactionState = RocketMQLocalTransactionState.ROLLBACK;
                break;
            case 1:
                rocketMQLocalTransactionState = RocketMQLocalTransactionState.COMMIT;
                break;
            case 2:
                rocketMQLocalTransactionState = RocketMQLocalTransactionState.ROLLBACK;
                break;
            default:
                rocketMQLocalTransactionState = RocketMQLocalTransactionState.ROLLBACK;
        }
        return rocketMQLocalTransactionState;

    }
}